package com.lvmama.rhino.analyze.push

import java.util.Date

import com.lvmama.rhino.common.entity.JobConfig
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.{DateFormatUtils, DateUtils}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions.{expr, round, _}


/**
  * @author wuchuantong
  * @version V1.0
  * @Title: todo
  * @Package com.lvmama.rhino.analyze.push
  * @Description: todo
  * @date 2017/4/5 16:33
  */
class UserClickAnalysis(config: JobConfig) extends SparkApplication {

  override var appName: String = "UserClickAnalysis"
  override var sparkConfig: Map[String, String] = config.spark

  val TBL_PUSH_MSG_TRACE: String = "push_msg_trace"
  val FAM_BASE_INFO: String = "base_info"
  val TBL_PUSH_USER_CLICK_ANALYSIS: String = "push_user_click_analysis"

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)
    //sparkConfig += ("spark.master" -> "local[2]")

    withSparkContext {
      sc =>
        val sqlContext = new SQLContext(sc)
        import com.lvmama.rhino.common.utils.hbase._
        import sqlContext.implicits._


        //取出设备及用户信息
        val keyLen = 28
        //取出180天内的信息
        val endRow = StringUtils.leftPad(DateFormatUtils.format(new Date(), "yyyyMMdd"), keyLen, "Z")
        val startRow = StringUtils.leftPad(DateFormatUtils.format(DateUtils.addDays(new Date(), -180), "yyyyMMdd"), keyLen, "0")

        //找出180内省推，会员模型，个性化的推送记录
        val msgTraceDF = sc.hbaseTable[(Option[String], Option[String], Option[String], Option[String])](TBL_PUSH_MSG_TRACE)
          .select("userId", "pushCategory", "click")
          .inColumnFamily(FAM_BASE_INFO)
          .withStartRow(startRow).withStopRow(endRow)
          .filter(x => x._1.isDefined && x._2.isDefined && x._3.isDefined && x._4.isDefined)
          .filter(f => (f._3.get.equals("NORMAL") || f._3.get.equals("MEMBER_MODEL") || f._3.get.equals("PERSONALISE")) && !f._2.get.equals("00000000"))
          .map(f => (f._1, f._2, f._3, f._4)).toDF("msgTraceRowKey", "userId", "pushCategory", "click")

        msgTraceDF.show()
        //用户180天内省推，会员模型，个性化和总的用户点击率
        msgTraceDF
          .groupBy(col("userId"))
          .agg(count("*").as("user180ReceiveCount"),
            sum(coalesce(col("click"), expr("0"))).as("user180ClickCount"),
            coalesce(sum(when(col("pushCategory") === "NORMAL", 1).otherwise(0)), expr("0")).as("province180ReceiveCount"),
            coalesce(sum(when(col("pushCategory") === "NORMAL" && col("click") === "1", 1).otherwise(0)), expr("0")).as("province180ClickCount"),
            coalesce(sum(when(col("pushCategory") === "MEMBER_MODEL", 1).otherwise(0)), expr("0")).as("member180ReceiveCount"),
            coalesce(sum(when(col("pushCategory") === "MEMBER_MODEL" && col("click") === "1", 1).otherwise(0)), expr("0")).as("member180ClickCount"),
            coalesce(sum(when(col("pushCategory") === "PERSONALISE", 1).otherwise(0)), expr("0")).as("personalise180ReceiveCount"),
            coalesce(sum(when(col("pushCategory") === "PERSONALISE" && col("click") === "1", 1).otherwise(0)), expr("0")).as("personalise180ClickCount"))
          .withColumn("province180ClickRate", coalesce(round(col("province180ClickCount") / col("province180ReceiveCount") * 100, 2), expr("0")))
          .withColumn("member180ClickRate", coalesce(round(col("province180ClickCount") / col("member180ReceiveCount") * 100, 2), expr("0")))
          .withColumn("personalise180ClickRate", coalesce(round(col("personalise180ClickCount") / col("personalise180ReceiveCount") * 100, 2), expr("0")))
          .withColumn("user180ClickRate", coalesce(round(col("user180ClickCount") / col("user180ReceiveCount") * 100, 2), expr("0")))
          .withColumn("userRowKey", rpad(reverse(col("userId")), 8, "0"))
          .map(m => (m.get(13).toString, m.get(0).toString, m.get(1).toString, m.get(2).toString, m.get(9).toString,
            m.get(3).toString, m.get(4).toString, m.get(10).toString,
            m.get(5).toString, m.get(6).toString, m.get(11).toString,
            m.get(7).toString, m.get(8).toString, m.get(12).toString))
          .rdd
          .toHBaseTable(TBL_PUSH_USER_CLICK_ANALYSIS)
          .toColumns("userId", "user180ReceiveCount", "user180ClickCount", "user180ClickRate",
            "province180ReceiveCount", "province180ClickCount", "province180ClickRate",
            "member180ReceiveCount", "member180ClickCount", "member180ClickRate",
            "personalise180ReceiveCount", "personalise180ClickCount", "personalise180ClickRate")
          .inColumnFamily(FAM_BASE_INFO).save()
    }

  }
}

object UserClickAnalysis {
  val config = JobConfig()

  def apply(): Unit = new UserClickAnalysis(config).execute()
}
